package com.jltx.rx;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

/**
 * @author jltxseo
 *         Created by junlintianxia on 2017/08/09.
 */

public class RxJava2Learn1 {

    public  static void main(String[] args){
        RxJava2Learn1 rxJava2Learn1 = new RxJava2Learn1();
//        rxJava2Learn1.createObservable();
//        rxJava2Learn1.createObservablePreComplete();
//        rxJava2Learn1.createObservablePreErro();
//        rxJava2Learn1.createObservableMutilErro();
//        rxJava2Learn1.createObservableDispose();
//        rxJava2Learn1.createFlowable();
//        rxJava2Learn1.createFlowableJust();
//        rxJava2Learn1.createObservableErroComplete();
//        rxJava2Learn1.createObservableCompleteErro();
        rxJava2Learn1.createObservableDisposeErro();
    }

    public void createObservable() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                System.out.println("onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    public void createObservablePreComplete() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 onComplete事件");
                e.onComplete();
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                System.out.println("onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    public void createObservablePreErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 onError");
                e.onError(new NullPointerException());
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                System.out.println("onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    /**
     * 演示发送多个error报错
     */
    public void createObservableMutilErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 onError事件");
                e.onError(new NullPointerException());
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                System.out.println("发射 onError事件");
                e.onError(new NullPointerException());
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                System.out.println("onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    public void createObservableDispose() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                System.out.println("发射 D");
                e.onNext("D");
                System.out.println("发射 onComplete事件");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            private Disposable disposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                i++;
                System.out.println(value);
                if (i == 2) {
                    System.out.println("disponsable---");
                    disposable.dispose();
                    System.out.println("isDisponsable:" + disposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    public void createFlowable(){

        // create a flowable
        Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("Hello Rx2");
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER);

        // create subscriber
        Subscriber subscriber = new Subscriber<String>(){
            @Override
            public void onSubscribe(Subscription s) {
                System.out.println("onSubscribe");
                //调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，可以写成Long.MAX_VALUE。
                // 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(String s) {
                System.out.println("onNext:"+s);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };

        flowable.subscribe(subscriber);
    }


    public void createFlowableJust(){
        Flowable<String> flowable = Flowable.just("Hello Rx2","ceshi");

        //仅仅关心onNext方法。所以可以简写成下面这样
        Consumer consumer = new Consumer<String>(){
            @Override
            public void accept(String s) throws Exception {
                System.out.println("accept:"+s);
            }
        };

        // create subscriber
        Subscriber subscriber = new Subscriber<String>(){
            @Override
            public void onSubscribe(Subscription s) {
                System.out.println("onSubscribe");
                //调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，可以写成Long.MAX_VALUE。
                // 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(String s) {
                System.out.println("onNext:"+s);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };

        flowable.subscribe(subscriber);
    }

    /**
     * 先发射erro接着发射complete,不会报错，但上游发射onErro事件后还可以继续发射，而下游接收了onErro事件之后并不会再继续接收
     */
    public  void createObservableErroComplete() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                System.out.println("发射 onError事件");
                e.onError(new NullPointerException());
                System.out.println("发射 D");
                e.onNext("D");
                System.out.println("发射 onComplete事件");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                System.out.println(value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    /**
     * 先发射complete接着发射erro,会报错
     */
    public  static void createObservableCompleteErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                System.out.println("发射 onComplete事件");
                e.onComplete();
                System.out.println("发射 D");
                e.onNext("D");
                System.out.println("发射 onError事件");
                e.onError(new NullPointerException());
            }
        }).subscribe(new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                System.out.println(value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }


    public void createObservableDisposeErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                System.out.println("发射 A");
                e.onNext("A");
                System.out.println("发射 B");
                e.onNext("B");
                System.out.println("发射 C");
                e.onNext("C");
                System.out.println("发射 D");
                e.onNext("D");
                System.out.println("发射 onError事件");
                e.onError(new NullPointerException());
            }
        }).subscribe(new Observer<String>() {
            private Disposable disposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(String value) {
                i++;
                System.out.println(value);
                if (i == 2) {
                    System.out.println("disponsable---");
                    disposable.dispose();
                    System.out.println("isDisponsable:" + disposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

}
